package cn.doitedu.demo2;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * 针对  活跃等级 为 C 的用户
 * 在规则上线后，如果这些用户 发生了 X行为 >=3 次
 * 等他再发生  Y行为的时候，触发营销消息
 {
      "rule_id":"rule-1",
      "static_profile_condition":{
            "tag_name":"tag0103",
            "tag_value":"C"
       },
      "realtime_profile_condition":{
            "event_id":"X",
            "min_count": 3
       },
      "trigger_condition":{
             "event_id":"Y"
       }
 }

 * 针对  价值等级 为 A 的用户                    ==>  静态画像条件
 * 在规则上线后，如果这些用户 发生了 W行为 >=2 次   ==>  动态画像条件
 * 等他再发生 K行为的时候，触发营销消息            ==>  触发条件
 {
 "rule_id":"rule-2",
 "static_profile_condition":{
 "tag_name":"tag0204",
 "tag_value":"A"
 },
 "realtime_profile_condition":{
 "event_id":"W",
 "min_count": 2
 },
 "trigger_condition":{
 "event_id":"K"
 }
 }

 *
 *
 */

public class RuleModel_1_Calculator implements RuleCalculator {

    ValueState<Integer> state;
    Table table;


    JSONObject paramObject;
    String ruleId;


    @Override
    public void init(RuntimeContext runtimeContext, String ruleParamJson ) throws IOException {
        state = runtimeContext.getState(new ValueStateDescriptor<Integer>("rule-1-cnt", Integer.class));

        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "doitedu:2181");

        Connection connection = ConnectionFactory.createConnection(conf);
        table = connection.getTable(TableName.valueOf("user_profile"));

        paramObject = JSON.parseObject(ruleParamJson);
        ruleId = paramObject.getString("rule_id");


    }

    @Override
    public void calc(UserEvent userEvent, Collector<Message> out) throws IOException {

        long userId = userEvent.getUser_id();
        String eventId = userEvent.getEvent_id();


        // 1. 判断当前收到的行为数据的行为人，是否满足规则中的 静态画像条件

        // 从规则参数中，取出静态画像条件所要查询到 标签名
        JSONObject staticProfileCondition = paramObject.getJSONObject("static_profile_condition");
        String tagName = staticProfileCondition.getString("tag_name");
        String tagValue = staticProfileCondition.getString("tag_value");


        Get get = new Get(Bytes.toBytes(userId));
        get.addColumn(Bytes.toBytes("f"), Bytes.toBytes(tagName));

        Result result = table.get(get);
        byte[] valueBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes(tagName));
        String value = Bytes.toString(valueBytes);

        if (!tagValue.equals(value)) return;


        // 2. 判断当前收到的行为事件，是否是规则中 动态画像条件要统计的事件，如果是，则执行统计逻辑
        JSONObject realtimeProfileCondition = paramObject.getJSONObject("realtime_profile_condition");
        String realTimeProfileEventId = realtimeProfileCondition.getString("event_id");
        int realTimeProfileMinCount = realtimeProfileCondition.getIntValue("min_count");

        if (realTimeProfileEventId.equals(eventId)) {
            int oldCount = state.value() == null ? 0 : state.value();
            state.update(oldCount + 1);
        }


        // 3. 判断当前收到的行为事件，是否是规则中 触发条件要求的事件；如果是，则判断该用户的动态画像条件是否已经满足，如满足，则输出消息
        JSONObject triggerCondition = paramObject.getJSONObject("trigger_condition");
        String triggerEventId = triggerCondition.getString("event_id");

        if (triggerEventId.equals(eventId)) {
            Integer nowCount = state.value();
            if (nowCount != null && nowCount >= realTimeProfileMinCount) {
                // 输出消息
                out.collect(new Message(userId,userEvent.getAction_time(),ruleId));
            }


        }

    }
}
